package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;
import javafx.scene.media.VideoTrack;

import java.io.IOException;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/03/03/14:20
 * @Description: 通过这个类表示虚拟主机  每个虚拟主机下面都管理这自己的交换机 队列 绑定 消息 数据
 * 同时提供API供上层使用
 * 针对virtualHost这个类,作为业务逻辑的整合类,就需要针对代码中抛出的异常进行处理
 */
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    //创建操作交换机锁对象
    private final Object exchangeLocker = new Object();

    //创建操作队列锁对象
    private final Object queueLocker = new Object();

    public VirtualHost(String name) {
        this.virtualHostName = name;
        //对于memoryDataCenter来说,不需要初始化,只需要把对象new出来就行
        //但是针对DiskDataCenter来说,则需要进行初始化操作, 建库建表和初始数据的设定
        diskDataCenter.init();
        //还需要针对硬盘的数据,进行恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    //创建交换机
    //如果交换机不存在,就创建,如果存在,就直接返回
    //返回值是boolean
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        //把交换机名称加上虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //1. 判定该交换机是否已经存在 ,直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    //该交换机已经存在
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                // 2.真正的构建交换机 先构建Exchange对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3. 需要把构造好的交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //4. 把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机对象创建完成! exchangeName=" + exchangeName);
                //上述逻辑先写硬盘,再写内存,目的就是因为硬盘更容易写失败,硬盘写失败了,内存就不写了
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机对象创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //1.先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除!");
                }
                //2.删除硬盘上的数据
                if (toDelete.isAutoDelete()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //3.删除内存上的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        // 把队列的名称给拼接上虚拟主机的名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //1. 判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                //2. 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //3. 写硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //4. 写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    //删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //1.先根据队列名称查询
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在无法删除!");
                }
                //2. 删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //3.删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    //创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //1. 判断当前绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        //已经存在
                        throw new Exception("[VirtualHost] binding 已经存在! queueName = " + queueName +
                                ",exchangeName = " + exchangeName);
                    }
                    //2.验证bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey = " + bindingKey);
                    }
                    //3.创建一个binding对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //4. 获取对应的交换机和队列,如果交换机或者队列不存在,那么这样的绑定无法创建
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName = " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName = " + exchangeName);
                    }
                    //5.把刚才创建好的binding对象插入到硬盘和内存
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName + ",queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName + ",queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    //解除绑定
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //1. 先获取绑定,判断是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName +",queueName=" + queueName);
                    }
                    //2.无论绑定是否持久化.都从硬盘删除一下
                    diskDataCenter.deleteBinding(binding);
                    //3.删除内存
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
           System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }


    //发送消息到指定的交换机/队列
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) {
        try {
            //转换交换机名称
            exchangeName = virtualHostName + exchangeName;
            //2. 检查routingKey是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey非法!routingKey = " + routingKey);
            }
            //3. 先根据交换机名称查到交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }
            //4.判定交换机类型 不同类型的发送方式存在区别
            if(exchange.getType() == ExchangeType.DIRECT) {
               //按照直接交换机的方式来转发
                //直接以routingKey 作为队列的名字,直接把消息写入对应的队列中  此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                //5.构造消息对象
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                //6. 查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName= " + queueName);
                }
                //7. 队列存在 向队列中写入消息
                sendMessage(queue,message);
            } else {
                //按照fanout和topic方法来转发
                //5. 找到该交换机关联的所有绑定 并且遍历这些绑定队列
                ConcurrentHashMap<String,Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry : bindingsMap.entrySet()) {
                    //1). 获取绑定对象,判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue =  memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null) {
                        //此处不抛出异常
                        System.out.println("[VirtualHost] basicPublish 发送消息时发现队列不存在! queueName = " +binding.getQueueName());
                        continue;
                    }
                    //2). 构造一个消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //3). 判定消息是否能转发给该队列
                    // 如果是 fanout 所有的绑定队列都要发送
                    // 如果是topic 还需要判定下,bindingKey和routingKey是否匹配
                    if(!router.route(exchange.getType(),binding,message)) {
                        continue;
                    }
                    //4).真正转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    //写入消息
    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此处发送消息就是把消息写入到硬盘和内存中
        //判定消息是否持久化
        int deliverMode = message.getDeliverMode();
        //1 不持久化   2 持久化
        if(deliverMode == 2) {
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);

        //通知消费者消费
        consumerManager.notifyConsume(queue.getName());
    }


    //订阅消息  添加一个队列的订阅者 当队列收到消息之后,就要把消息交给对应的订阅者
    //consumerTag: 表示消费者的身份标识,
    //autoAck:消息被消费完成后应答的方式  true为自动应答,false为手动应答
    //consumer为函数式接口 回调函数
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //构造一个ConsumerEnv对象,把这个对应的队列找到,再把这个ConsumerEnv对象添加到该队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    //消息确认
    public boolean basicAck(String queueName,String messageId) {
        queueName = virtualHostName + queueName;
        try{
            //1. 获取到消息和队列
           Message message = memoryDataCenter.getMessage(messageId);
           if(message == null) {
               throw new MqException("[VirtualHost] 要确认的消息不存在 messageId=" + messageId);
           }
           MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在 queueName=" + queueName);
            }
            //2. 删除硬盘上的数据
            if(message.getDeliverMode() ==2) {
                diskDataCenter.deleteMessage(queue,message);
            }
            //3. 删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            //4. 删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName + ",messageId=" + messageId);
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败! queueName=" + queueName + ",messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
